package com.atguigu.day07;

import com.atguigu.bean.Event;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;

import java.util.Calendar;
import java.util.Random;

public class Flink02_SplitStreamByFilter {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.从自定义数据源获取数据
        DataStreamSource<Event> streamSource = env.addSource(new ClickSource());

        //我们可以将电商网站收集到的用户行为数据进行一个拆分，根据类型（type）的不同，分为“Songsong”的浏览数据、“Bingbing”的浏览数据等。
        SingleOutputStreamOperator<Event> songsongStream = streamSource.filter(new FilterFunction<Event>() {
            @Override
            public boolean filter(Event value) throws Exception {
                return "Songsong".equals(value.user);
            }
        });

        SingleOutputStreamOperator<Event> bingbingStream = streamSource.filter(new FilterFunction<Event>() {
            @Override
            public boolean filter(Event value) throws Exception {
                return "Bingbing".equals(value.user);
            }
        });

        SingleOutputStreamOperator<Event> otherStream = streamSource.filter(new FilterFunction<Event>() {
            @Override
            public boolean filter(Event value) throws Exception {
                return !("Songsong".equals(value.user) || "Bingbing".equals(value.user));
            }
        });

        songsongStream.print("song");
        bingbingStream.print("bing");
        otherStream.print("other");

        env.execute();





    }
    // 这里重新实现了ClickSource()，因为我们在自定义数据源中发送了水位线
    public static class ClickSource implements SourceFunction<Event> {

        private boolean running = true;

        @Override
        public void run(SourceContext<Event> sourceContext) throws Exception {

            Random random = new Random();
            String[] userArr = {"Songsong", "Bingbing", "Lili"};
            String[] urlArr  = {"./home", "./cart", "./prod?id=1"};

            while (running) {
                long currTs = Calendar.getInstance().getTimeInMillis(); // 毫秒时间戳
                String username = userArr[random.nextInt(userArr.length)];
                String url      = urlArr[random.nextInt(urlArr.length)];
                Event event = new Event(username, url, currTs);

                // 使用collectWithTimestamp方法将数据发送出去，并指明数据中的时间戳的字段
                sourceContext.collectWithTimestamp(event, event.timestamp);

                // 发送水位线
                sourceContext.emitWatermark(new Watermark(event.timestamp - 1L));
                Thread.sleep(1000L);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }

}
